package org.xiaoxin.onecoupon.framework.idempotent;

import cn.hutool.extra.expression.engine.spel.SpELEngine;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.Signature;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.data.redis.core.PartialUpdate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.RedisScript;
import org.xiaoxin.onecoupon.framework.exception.ServiceException;

import java.lang.reflect.Method;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * 防止消费队列消费者重复消费消息切面控制器
 * @author xiaoxin
 * 2025/5/9
 */
@Aspect
@Slf4j
@RequiredArgsConstructor
public class NoMQDuplicateConsumeAspect {

   private final StringRedisTemplate stringRedisTemplate;
   /**
    *
    */
   private static final String LUA_SCRIPT = """
           local key = KEYS[1]
           local value = ARGV[1]
           local expire_time_ms = ARGV[2]                  
           -- 获取旧值
           local old_value = redis.call('GET', key)         
           -- 如果键不存在，则设置新值并设置过期时间
           if not old_value then
               redis.call('SET', key, value, 'PX', expire_time_ms)
           end        
           return old_value
            """;
   @Around("@annotation(org.xiaoxin.onecoupon.framework.idempotent.NoMQDuplicateConsume)")
   public Object noMQRepeatConsume(ProceedingJoinPoint joinPoint) throws Throwable{
      //获取注解实例
      NoMQDuplicateConsume noMQDuplicateConsume=getNoMQDuplicateConsumeAnnotation(joinPoint);
      String uniqueKey= noMQDuplicateConsume.keyPrefix()+ SpELUtil.parseKey(noMQDuplicateConsume.key(),
              ((MethodSignature) joinPoint.getSignature()).getMethod(),joinPoint.getArgs());
      //执行lua脚本
      /**
       *     absentAndGet 为空：代表消息是第一次到达，执行完 LUA 脚本后，会在 Redis 设置 Key 的 Value 值为 0，消费中状态。
       *     absentAndGet 为 0：代表已经有相同消息到达并且还没有处理完，会通过抛异常的形式让 RocketMQ 重试。
       *     absentAndGet 为 1：代表已经有相同消息消费完成，返回空表示不执行任何处理。
       */
         String absentAndGet = stringRedisTemplate.execute(
              RedisScript.of(LUA_SCRIPT, String.class),
              List.of(uniqueKey),
              IdempotentMQConsumeStatusEnum.CONSUMING.getCode(),
              String.valueOf(TimeUnit.SECONDS.toMillis(noMQDuplicateConsume.keyTimeout()))
      );
      if (Objects.nonNull(absentAndGet)){
         boolean errorFlag = IdempotentMQConsumeStatusEnum.isError(absentAndGet);
         log.warn("[{}] MQ repeated consumption, {}.", uniqueKey, errorFlag ? "Wait for the client to delay consumption" : "Status is completed");
         if (errorFlag){
            throw new ServiceException(String.format("消息消费者幂等异常，幂等标识：%s", uniqueKey));
         }
         return  null;
      }
      Object result;
      try {
         //执行标记了防消费队列重新消费消息的方法原逻辑
         result=joinPoint.proceed();
         // 设置幂等标记为已消费并设置防重令牌 Key 过期时间，单位秒
         stringRedisTemplate.opsForValue().set(uniqueKey, IdempotentMQConsumeStatusEnum.CONSUMED.getCode(), noMQDuplicateConsume.keyTimeout(), TimeUnit.SECONDS);
      }catch (Throwable ex){
         // 删除幂等 Key，让消息队列消费者重试逻辑进行重新消费
         stringRedisTemplate.delete(uniqueKey);
         throw ex;
      }
      return  result;

   }

   /**
    * 返回自定义防重复注解实例
    * @param joinPoint
    * @return
    */
   private NoMQDuplicateConsume getNoMQDuplicateConsumeAnnotation(ProceedingJoinPoint joinPoint) throws NoSuchMethodException{
      MethodSignature methodSignature = (MethodSignature) joinPoint.getSignature();
      Method declaredMethod = joinPoint.getTarget().getClass().getDeclaredMethod(methodSignature.getName(), methodSignature.getMethod().getParameterTypes());
      return  declaredMethod.getAnnotation(NoMQDuplicateConsume.class);
   }

}
